package xj.toolkit.netty.transport.endpoint;

import org.apache.commons.pool2.ObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import xj.toolkit.base.Sender;
import xj.toolkit.netty.transport.PoolableSender;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

/**
 * 使用时可以按需创建连接并存放在连接池中的Endpoint
 * 
 * @param <I>
 */
public class PoolableEndpoint<I> implements Sender<I> {
	private static Logger logger = LoggerFactory.getLogger(PoolableEndpoint.class);

	ObjectPool<PoolableSender<I>> pool;
	private long waitTimeout = 1000;
	private BlockingQueue<I> sendQueue = null;

	private ExecutorService exec = Executors.newSingleThreadExecutor();

	@PostConstruct
	public void init() {
		doSendPending();
	}

	public void send(I bean) {
		if (null != this.sendQueue) {
			if (logger.isTraceEnabled()) {
				logger.trace("send bean:" + bean);
			}
			sendQueue.add(bean);
		} else {
			logger.warn("Unsupport Operation for send.");
		}
	}

	public void stop() {
		if (null != this.sendQueue) {
			this.sendQueue.clear();
		}
	}

	private void doSendPending() {
		exec.submit(new Runnable() {

			public void run() {
				sendPending();
			}
		});
	}

	private void sendPending() {
		try {
			I bean = sendQueue.poll(waitTimeout, TimeUnit.MILLISECONDS);
			if (null != bean) {
				PoolableSender<I> client = null;
				for (int i = 0; i < 5; i++) {
					client = pool.borrowObject();
					client.setReturnedToPool(false);
					logger.debug("borrowed {} from pool", client);
					if (client.getChannel() == null || !client.getChannel().isActive() && !client.isReturnedToPool()) {
						logger.debug("client {} is not connected,drop it to pool", client);
						client.setReturnedToPool(true);
						pool.returnObject(client);
					} else {
						break;
					}
				}
				if (client != null) {
					logger.debug("use client {} to send msg {}", client, bean);
					client.send(bean);
				} else {
					logger.error("can't obtain available client from pool");
				}
			}
		} catch (Exception e) {
			logger.error("sendPending error:", e);
		} finally {
			doSendPending();
		}
	}

	public int getPendingCount() {
		if (null != this.sendQueue) {
			return this.sendQueue.size();
		} else {
			return -1;
		}
	}

	public void setWaitTimeout(long waitTimeout) {
		this.waitTimeout = waitTimeout;
	}

	public void setSendQueue(BlockingQueue<I> sendQueue) {
		this.sendQueue = sendQueue;
	}

	public void setPool(ObjectPool<PoolableSender<I>> pool) {
		this.pool = pool;
	}

	@Override
	public void send(Object req, I response) {
		this.send(response);
	}

	@Override
	public boolean prepared() {
		return pool != null && (pool.getNumActive() + pool.getNumIdle()) > 0;
	}
}
